package rabbitmq

import (
	"errors"
	"fmt"
	"github.com/streadway/amqp"
	"math/rand"
	"strings"
	"sync"
	"time"
)

var RabbitPoolServe *RabbitPool

// rabbitmq线程池
type RabbitPool struct {
	AMQPURL string
	ConnectionNum int  // 连接数
	ChannelNum int // channel数
	// Pool map[int]*Pool // 获取当个pool对象

	Connections map[int]*amqp.Connection
	Channels map[int]Channel
	ChannelConectionMap map[int]int // 频道和channel key和connection映射
	IdelChannel []int // 空闲channel
	BusyChannel map[int]int // 繁忙channel
	m *sync.Mutex
}

type Pool struct {
	Chl *amqp.Channel
	ChannelId int // 属于pool那个channel
	ExName string // 交换机名称
	RouteKey string // routekey
	QName string // 队列名
}

type Channel struct {
	conn *amqp.Channel
}

func(r *RabbitPool) InitRabbit() {
	if r.ConnectionNum == 0 {
		r.ConnectionNum = 1
	}
	if r.ChannelNum == 0 {
		r.ChannelNum = 10
	}
	r.Connections = make(map[int]*amqp.Connection, r.ConnectionNum)
	r.Channels = make(map[int]Channel, r.ChannelNum)
	r.BusyChannel = make(map[int]int, r.ChannelNum)
	r.ChannelConectionMap = make(map[int]int, r.ChannelNum)
	// r.Pool = make(map[int]*Pool, r.ChannelNum)
	r.m = new (sync.Mutex)

	r.createConnectPool()
	r.createChannelPool()

	// 定时回收每个channel

}

func failOnError(err error) {
	if err != nil {
		fmt.Println(err.Error())
	}
}

func(r *RabbitPool) createConnect() *amqp.Connection{
	conn, err := amqp.Dial(r.AMQPURL)
	failOnError(err)
	return conn
}

func(r *RabbitPool) createConnectPool() {
	for i:=0; i<r.ConnectionNum; i++ {
		r.Connections[i] = r.createConnect()
	}
}

func(r *RabbitPool) createChannel(channelKey int, conn *amqp.Connection) Channel{
	// 判断connection是否已经关闭
	if conn.IsClosed() {
		// 重新创建connection
		conn = r.createConnect()
		r.m.Lock()
		defer r.m.Unlock()
		index := r.ChannelConectionMap[channelKey]
		r.Connections[index] = conn
	}
	ch := Channel{
		conn: nil,
	}
	chl, err := conn.Channel()
	if err != nil {
		chl = r.recreateChannel(channelKey, conn, err)
	}
	// 开启confirm 模式
	chl.Confirm(false)
	ch.conn = chl

	return ch
}

func (r *RabbitPool) recreateChannel(channelKey int, connection *amqp.Connection, err error)(ch *amqp.Channel){
	if strings.Index(err.Error(),"channel/connection is not open") >= 0 || strings.Index(err.Error(),"CHANNEL_ERROR - expected 'channel.open'") >=0{
		ch, err = connection.Channel()
		failOnError(err)
	}else{
		failOnError(err)
	}
	return
}

func(r * RabbitPool) createChannelPool() {
	for k,v := range r.Connections {
		for j:=0; j<r.ChannelNum ; j++ {
			key := k*r.ChannelNum+j
			r.ChannelConectionMap[key] = k
			tempc := r.createChannel(key, v)
			r.Channels[key] = tempc
			r.IdelChannel = append(r.IdelChannel, key)
		}
	}
}

func (r *RabbitPool) GetPool() (*Pool, error) {
	r.m.Lock()
	// defer  r.m.Unlock()
	leng := len(r.IdelChannel)
	if leng > 0 {
		// 随机种子，不设置时间戳，每次随机不会改变
		rand.Seed(time.Now().Unix())
		index := rand.Intn(leng)
		channelId := r.IdelChannel[index]
		r.IdelChannel = append(r.IdelChannel[:index], r.IdelChannel[index+1:]...)
		r.BusyChannel[channelId] = channelId
		ch := r.Channels[channelId].conn
		tempPool := &Pool{
			Chl:      ch,
			ChannelId: channelId,
			ExName:   "",
			RouteKey: "",
			QName:    "",
		}
		r.m.Unlock()
		return tempPool, nil
	}else {
		return nil, errors.New("pool已经全部使用中...")
	}
}

func(p *Pool) CreateExTopic(exName string, routerKey string) {
	err := p.Chl.ExchangeDeclare(
		exName, // name
		"topic",      // type
		true,         // durable
		false,        // auto-deleted
		false,        // internal
		false,        // no-wait
		nil,          // arguments
	)
	p.ExName = exName
	p.RouteKey = routerKey
	failOnError(err)
}

func (p * Pool) CreateQueuen(qName string) {
	p.Chl.QueueDeclare(
		qName,
		true,
		false,
		false,
		false,
		nil,
	)
	p.Chl.QueueBind(
		qName,
		p.RouteKey,
		p.ExName,
		false,
		nil,
	)
	p.QName = qName
}

func (r *RabbitPool) backChannelId(channelId int) {
	r.m.Lock()
	delete(r.BusyChannel, channelId)
	r.IdelChannel = append(r.IdelChannel, channelId)
	r.m.Unlock()
}

func (r *RabbitPool) ReChannelJoin(channelId int) {
	r.m.Lock()
	tempc := r.createChannel(channelId, r.Connections[r.ChannelConectionMap[channelId]])
	r.Channels[channelId] = tempc
	delete(r.BusyChannel, channelId)
	r.IdelChannel = append(r.IdelChannel, channelId)
	r.m.Unlock()
}

func (p *Pool) PublicMSg(msg string, r *RabbitPool) error {
	confirm := make(chan amqp.Confirmation, 100)
	rconfirm := p.Chl.NotifyPublish(confirm)
	t := time.NewTicker(time.Second * 5)
	// for {
		err := p.Chl.Publish(
			p.ExName,
			p.RouteKey,
			false,
			false,
			amqp.Publishing{
				DeliveryMode: amqp.Persistent,
				ContentType: "text/plain",
				Body:        []byte(msg),
			},
		)
		if err != nil {
			//出现错误，判断是否是channel/connection colse错误
			return err
		}
		select {
		case rf := <-rconfirm:
			fmt.Println("ack: ",rf.DeliveryTag, rf.Ack)
			if rf.Ack == true && rf.DeliveryTag > 0 {
				// 创建新线程加入到idle中
				r.ReChannelJoin(p.ChannelId)
				p.Chl.Ack(rf.DeliveryTag, false)
			}else if rf.Ack == false && rf.DeliveryTag > 0 {
				p.Chl.Nack(rf.DeliveryTag, false, false)
			}
		case <-t.C:
			// 定时15秒，回收channel
			r.ReChannelJoin(p.ChannelId)
			t.Stop()
			break
		}
	// }

	return nil
}

func (p * Pool) CreateCustmer(name string, qos int) (<-chan amqp.Delivery,error){
	p.Chl.Qos(qos,0,false)  // 限制每次获取的条数，不ack每次只取一条
	msgs, err := p.Chl.Consume(p.QName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
	return msgs, err
}
